package com.atbeijing.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

object SparkStreaming04_Source_DIY {

    def main(args: Array[String]): Unit = {

        // TODO 创建环境对象
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")

        // 构建StreamingContext对象时，第二个参数表示数据 采集周期，以毫秒单位，一般以秒为单位使用
        val ssc = new StreamingContext(sparkConf, Seconds(3))

        val recDS: ReceiverInputDStream[String] = ssc.receiverStream(new MyReceiver())

        recDS.print()

        ssc.start()
        ssc.awaitTermination()
    }
    // 自定义数据采集器
    // 1. 继承Receiver
    // 2. 传递父类参数以及定义泛型
    // 3. 重写方法
    class MyReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY){
        private var flg = true
        // 采集数据
        override def onStart(): Unit = {

            while ( flg ) {
                // 生成数据
                val name = "zhangsan-" + System.currentTimeMillis()
                // 将数据交给采集器管理
                store(name)
                Thread.sleep(1000)
            }
        }

        // 释放资源
        override def onStop(): Unit = {
            flg = false
        }
    }
}
